package com.galeno.day04;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.triggers.CountTrigger;
import org.apache.flink.streaming.api.windowing.triggers.PurgingTrigger;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;

import java.sql.DriverManager;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/2019:50
 */
public class CountWindowAllDemo {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",55555);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        DataStreamSource<String> lines = env.socketTextStream("192.168.77.101", 8888);
        SingleOutputStreamOperator<Integer> numed = lines.map(Integer::parseInt);
        //AllWindowedStream<Integer, GlobalWindow> countWindowAll = numed.countWindowAll(2);
        AllWindowedStream<Integer, GlobalWindow> countWindowAll = numed.windowAll(GlobalWindows.create()).trigger(PurgingTrigger.of(CountTrigger.of(5)));
        SingleOutputStreamOperator<Integer> sumed = countWindowAll.sum(0);
        sumed.print();
        env.execute();


    }
}
//class MyEnv implements StreamExecutionEnvironmentFactory{
//
//    @Override
//    public StreamExecutionEnvironment createExecutionEnvironment(Configuration configuration) {
//        return StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
//    }
//}
